package com.nl.utils.aggfunctions;

import com.nl.bean.input.UserBehavior;
import java.math.BigDecimal;
import java.math.RoundingMode;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;

/**
 *Description:自定义预聚合函数实现计算平均值
 *            AggregateFunction<IN,ACC,OUT>
 *            IN:输入的数据类型
 *            ACC:中间的聚合状态的数据类型
 *            OUT:输出的数据类型,也是后面的聚合结果输出函数的输入
 *Version:1.0.0
 *@author shihb
 *@date 2020/4/3 15:32
 */
public class AverageAgg<IN> implements AggregateFunction<IN, Tuple2<Long,Long>,Double> {

  @Override
  public Tuple2<Long, Long> createAccumulator() {
    // 初始化累加器
    return new Tuple2<Long, Long>(0L,0L);
  }

  @Override
  public Tuple2<Long, Long> add(IN in, Tuple2<Long, Long> acc) {
    // 获取求和的总数sum和计算count
    Long sum =0L;
    Long count=0L;
    if(UserBehavior.class.isInstance(in)){
      UserBehavior ub=(UserBehavior)in;
      sum=ub.getTimestamp()+acc.f0;
      count=acc.f1+1;
    }
    return new Tuple2<Long, Long>(sum,count);
  }

  @Override
  public Double getResult(Tuple2<Long, Long> acc) {
    //获取平均值
    // BigDecimal财务上用的计算
    BigDecimal sum = new BigDecimal(acc.f0);
    BigDecimal count = new BigDecimal(acc.f1);
    // sum/count 4舍5入 保存两位小数
    BigDecimal result = sum.divide(count,2, RoundingMode.HALF_UP);
    return result.doubleValue();
  }

  @Override
  public Tuple2<Long, Long> merge(Tuple2<Long, Long> acc1, Tuple2<Long, Long> acc2) {
    // 两个累加器怎么处理
    Long sum =acc1.f0+acc2.f0;
    Long count=acc1.f1+acc2.f1;
    return new Tuple2<Long, Long>(sum,count);
  }
}
